package com.jnetx.javax.sip.trans;

import com.jnetx.javax.sip.TransportManager;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import javax.sip.ListeningPoint;
import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.channels.spi.SelectorProvider;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

/**
 * Created by IntelliJ IDEA.
 * User: deps
 * Date: 28.03.11
 * Time: 20:29
 * To change this template use File | Settings | File Templates.
 */
public class TCPTransportManager extends Thread implements TransportManager, TransportListener {
    private final static Log log = LogFactory.getLog(TCPTransportManager.class);

    private ServerSocketChannel serverChannel;
    private Selector selector;
    private Map<SocketChannel, ChannelReader> readers = new HashMap<SocketChannel, ChannelReader>();
    private final ByteBuffer readBuffer = ByteBuffer.allocate(8192);
    private final byte[] readBufferBack = new byte[8192];

    public TCPTransportManager() {
        super("TCPTransportManager");
        try {
            selector = SelectorProvider.provider().openSelector();
        } catch (IOException e) {
            log.error("Cannot get selector!", e);
        }
        start();
    }

    public void onMessageReceived() {

    }

    public void startListeningPoint(ListeningPoint listeningPoint) {
        try {
            serverChannel = ServerSocketChannel.open();
            serverChannel.configureBlocking(false);
            InetSocketAddress isa = new InetSocketAddress(listeningPoint.getIPAddress(), listeningPoint.getPort());
            serverChannel.socket().bind(isa);
            serverChannel.register(selector, SelectionKey.OP_ACCEPT);
        } catch(IOException e) {
            log.error("Cannot start the TCP listening point on "+listeningPoint.getIPAddress()+":"+listeningPoint.getPort());
        }
    }

    @Override
    public void run() {
        if (log.isDebugEnabled()) log.info("Starting TCP transport, no listeners yet...");
        while (!isInterrupted()) {
            try {
                selector.select();
                Iterator selectedKeys = selector.selectedKeys().iterator();
                while (selectedKeys.hasNext()) {
                    SelectionKey key = (SelectionKey) selectedKeys.next();
                    selectedKeys.remove();
                    if (!key.isValid()) continue;
                    if (key.isAcceptable()) {
                        accept(key);
                    }
                }
                selector.selectedKeys().clear();
            } catch (Exception ex) {
                log.error("Error during accepting of incoming connections!", ex);
            }
        }
        if (log.isDebugEnabled()) log.info("TCP transport has been stopped.");
    }

    private void accept(SelectionKey key) throws IOException {
        ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();
        SocketChannel channel = serverChannel.accept();
        // set the new channel blocking
        channel.configureBlocking(true);
        ChannelReader reader = new ChannelReader(channel, this);
        readers.put(channel, reader);
        reader.start();
        if (log.isDebugEnabled()) log.debug("Connection has been accepted.");
    }

    private static int indexSeed = 1;

    private class ChannelReader extends Thread {
        private ObjectInputStream stream;
        private TransportListener listener;

        private ChannelReader(SocketChannel channel, TransportListener listener) throws IOException {
            super("SIP.TCP-reader-" + indexSeed++);
            setDaemon(true);
            this.stream = new ObjectInputStream(new BufferedInputStream(Channels.newInputStream(channel)));
            this.listener = listener;
        }

        @Override
        public void run() {
            if (log.isDebugEnabled()) log.info(getName() + " has been started.");
//            LogSource logSource = null;
//            //Reading log source information
//            try {
//                OpenEvent openEvent = (OpenEvent) stream.readObject();
//                logSource = openEvent.getLogSource();
//                logSourceListener.logSourceAttached(logSource);
//                if (logger.isDebugEnabled()) logger.debug("Log Source Info: " + logSource);
//            } catch (Exception e) {
//                logger.error("Cannot read log source info from the input stream!", e);
//            }
//            if (logSource != null) {
//                while (!isInterrupted()) {
//                    try {
//                        Object obj = stream.readObject();
//                        if (obj instanceof CloseEvent) break;
//                        Message message = (Message) obj;
//                        if (logger.isDebugEnabled()) logger.debug("message has been read: " + message);
//                        LogServer.this.messageReceived(logSource, message);
//                    } catch (Exception e) {
//                        logger.error("Cannot read from the input stream!", e);
//                        break;
//                    }
//                }
//            }
//            try {
//                stream.close();
//            } catch (IOException e) {
//            }
            if (log.isDebugEnabled()) log.info(getName() + " has been stopped.");
        }
    }

}
